package com.junqi.msg.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.junqi.common.entity.RedisKeyManager;
import com.junqi.msg.adapter.WSAdapter;
import com.junqi.msg.config.ThreadPoolConfig;
import com.junqi.msg.event.UserOfflineEvent;
import com.junqi.msg.event.UserOnlineEvent;
import com.junqi.msg.event.UserPullMsgEvent;
import com.junqi.msg.service.WebSocketService;
import com.junqi.msg.utils.JsonUtils;
import com.junqi.msg.utils.RedisUtil;
import com.junqi.msg.websocket.NettyUtil;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class WebSocketServiceImpl implements WebSocketService {

    @Autowired
    @Qualifier(ThreadPoolConfig.WS_EXECUTOR)
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;
    @Autowired
    private WSAdapter wsAdapter;

    /**
     * 所有已连接的websocket连接链接用户
     */
    private static final ConcurrentHashMap<String, Channel> userChannels = new ConcurrentHashMap<>();

    @Override
    public void connect(Channel channel) {
        String uid = NettyUtil.getAttr(channel, NettyUtil.UID);
        String roomId = NettyUtil.getAttr(channel, NettyUtil.ROOMID);
        userChannels.putIfAbsent(uid,channel);
        //用户上线事件
        applicationEventPublisher.publishEvent(new UserOnlineEvent(this, uid,roomId));
    }

    @Override
    public void removed(Channel channel) {
        String uid = NettyUtil.getAttr(channel, NettyUtil.UID);
        String roomId = NettyUtil.getAttr(channel, NettyUtil.ROOMID);
        userChannels.remove(uid);
        //用户下线事件
        applicationEventPublisher.publishEvent(new UserOfflineEvent(this, uid,roomId));
    }

    /**
     * 用户拉取消息
     */
    @Override
    public void userPuLLMsg(Channel channel, Long data) {
        String uid = NettyUtil.getAttr(channel, NettyUtil.UID);
        String roomId = NettyUtil.getAttr(channel, NettyUtil.ROOMID);
        //调用前端拉取消息事件
        applicationEventPublisher.publishEvent(new UserPullMsgEvent(this, roomId, data, uid));
    }

    /**
     * 用户查询房间人数
     */
    @Override
    public void roomCountMsg(Channel channel) {
        String uid = NettyUtil.getAttr(channel, NettyUtil.UID);
        String roomId = NettyUtil.getAttr(channel, NettyUtil.ROOMID);

        //获取房间缓存中的用户
        Set<String> userSet = RedisUtil.sGet(RedisKeyManager.generateKey(RedisKeyManager.ROOM_USERS, roomId));
        sendMessageToUid(JsonUtils.toStr(wsAdapter.buildRoomCountMsgResp(userSet.size())),uid);
    }


    /**
     * 给指定的一批用户（某个房间的用户）发消息
     * @param message
     * @param uidList
     */
    @Override
    public void sendMessageToUids(String message, Set<String> uidList) {
        if(StrUtil.isNotBlank(message) && CollUtil.isNotEmpty(uidList)){
            for (String uid : uidList) {
                threadPoolTaskExecutor.execute(() ->{
                    Channel channel = userChannels.get(uid);
                    if (channel == null || !channel.isActive()) {
                        //log.info("用户：{}不在线", uid);
                        return;
                    }
                    channel.writeAndFlush(new TextWebSocketFrame(message));
                });
            }
        }
    }

    /**
     * 给指定的用户发消息
     * @param message
     * @param uid
     */
    @Override
    public void sendMessageToUid(String message, String uid) {
        if(StrUtil.isNotBlank(message) && StrUtil.isNotBlank(uid)) {
            threadPoolTaskExecutor.execute(() -> {
                Channel channel = userChannels.get(uid);
                if (channel == null || !channel.isActive()) {
                    //log.info("用户：{}不在线", uid);
                    return;
                }
                channel.writeAndFlush(new TextWebSocketFrame(message));
            });
        }
    }

}
